
#pragma once

#include <ttg/threads.h>
#include <ttg/services.h>
#include <ttg/pipeline.h>
#include "Common.h"

using namespace ttg::math;
using namespace ttg::pipeline;

//Splitting range into parts to achieve good CPU usage.
class ttgSplitter :public Node<uint64, std::pair<uint64, uint64>>
{
	public:
		virtual void process(uint64 data)
		{
			size_t part_count = ttg::services::CPUAnalyzer::getRef()->getCoreCount() * 4;
			for (size_t i = 0; i < part_count; i++)
			{
				uint64 lo = (i * data) / part_count;
				uint64 hi = ((i + 1) * data) / part_count;
				if (lo != hi)
					sendNext(std::make_pair(lo, hi));
			}
		}
};

//Counts prime numbers in the required range.
class ttgPrimeCounter :public Node<std::pair<uint64, uint64>, uint64>
{
	private:
		PrimeFinderPool pool;
	public:
		//Counting.
		virtual void process(std::pair<uint64, uint64> data)
		{
			double t_start = ttg::services::Timer::getRef()->getSeconds();
			sendNext(pool.findPrimes(data.first, data.second));
			double t_end = ttg::services::Timer::getRef()->getSeconds();
		}
		//Returning difficulty of data processing (in arbitrary measure units).
		double getDifficulty(std::pair<uint64, uint64> data)
		{
			return data.second * log((long double)data.second / 3) -
				   data.first * log((long double)data.first / 3);
		}
};

//Gathering incoming prine count.
class ttgGatherer :public Node<uint64, uint64>
{
	private:
		ttg::threads::AtomicInt value;
	public:
		virtual void onBegin()
		{ value.set(0); }

		virtual void process(uint64 data)
		{ value.add((int)data); }

		virtual void onEnd()
		{sendNext(value.get()); }
};

std::pair<uint64, double> ttgPrimes(uint64 n)
{
	//Creating pipeline.
	static
	Pipeline<uint64, uint64> pl(wrap(new ttgSplitter()) +
								wrap(new ttgPrimeCounter()) +
								wrap(new ttgGatherer()));
	//Counting.
	double t_start = ttg::services::Timer::getRef()->getSeconds();
	uint64 count = pl.run(n + 1)[0];
	double t_end = ttg::services::Timer::getRef()->getSeconds();
	//Returning result.
	return std::make_pair(count, t_end - t_start);
}